//
// Created by connection on 17-10-1.
//


#include <android/looper.h>
#include <errno.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <sys/socket.h>
#include <sys/un.h>
#include <sys/un.h>
#include <sys/select.h>
#include <sys/types.h>
#include <pthread.h>
#include <semaphore.h>

#include "rocky_connection.h"
#include "../utils/connection_utils.h"
#include "../utils/app_sys_log.h"
#include "../utils/record_stream.h"
#include "../works/work_item_queue.h"
#include "../utils/write_data_record.h"

struct RockyConnection{
    int fd;
    int buf_size;
    int shutdown;
    sem_t sem;
    ALooper* client_looper;
    RecordStream* read_record_stream;
    WorkItemQueue* write_data_queue;
    on_received_data on_received;
    on_connection_disconnected on_disconnected;
};

static int rocky_connection_event_handler(int fd, int events, void* data);
static void rocky_connection_handle_write(RockyConnection* thiz);

RockyConnection* rocky_connection_create(int fd, size_t buf_size, on_received_data on_received,
                                 on_connection_disconnected on_disconnected) {
    RockyConnection* thiz = (RockyConnection *)malloc(sizeof(RockyConnection));
    thiz->fd = fd;
    thiz->shutdown = 0;
    thiz->buf_size = buf_size;
    thiz->on_received = on_received;
    thiz->on_disconnected = on_disconnected;
    thiz->read_record_stream = record_stream_new(thiz->fd, buf_size);
    thiz->write_data_queue = work_item_queue_create();
    sem_init(&thiz->sem, 0, 0);

    return thiz;
}

static void rocky_connection_handler_read(RockyConnection* thiz) {
    record_stream_read(thiz->read_record_stream);
    void* record_data = NULL;
    size_t record_size = 0;
    int ret = -1;
    while (1){
        ret = record_stream_get_next(thiz->read_record_stream, &record_data, &record_size);
        if(ret < 0 || (ret == 0 && record_data == NULL)) {
            break;
        }
        thiz->on_received(thiz, record_data, record_size);
    }
}

void rocky_connection_write_data(RockyConnection* thiz, void* data, size_t data_size) {
    WriteDataRecord* write_data = write_data_record_create(thiz->fd, data, data_size);
    work_item_queue_add(thiz->write_data_queue, write_data, write_data_record_destroy);
    rocky_connection_handle_write(thiz);
}
static void rocky_connection_handle_write(RockyConnection* thiz) {
    while (1) {
        WriteDataRecord *write_data = (WriteDataRecord *) work_item_queue_peek(
                thiz->write_data_queue);
        if (write_data == NULL) {
            //app_sys_error("%s %p, no write data", __func__, thiz);
            break;
        }
        int write_count = write_data_record_write(write_data);
        if(write_count >= 0) {
            if(write_data_record_finished(write_data)) {
                //全部写完，从队列中清除
                work_item_queue_remove(thiz->write_data_queue);
                write_data_record_destroy(write_data);
            }
        } else if(write_count < 0 && errno == EAGAIN) {
            ALooper_addFd(thiz->client_looper, thiz->fd,
                          0, ALOOPER_EVENT_INPUT | ALOOPER_EVENT_HANGUP | ALOOPER_EVENT_OUTPUT,
                          (ALooper_callbackFunc) rocky_connection_event_handler, thiz);
            break;
        } else {
            //TODO
            break;
        }
    }
}

static int rocky_connection_event_handler(int fd, int events, void* data) {
    RockyConnection* thiz = (RockyConnection *)data;
    if(events & (ALOOPER_EVENT_INPUT | ALOOPER_EVENT_ERROR)) {
        //app_sys_debug("%s handler read", __func__);
        rocky_connection_handler_read(thiz);
    } else if(events & ALOOPER_EVENT_HANGUP) {
        //app_sys_debug("%s handler disconnect", __func__);
        thiz->on_disconnected(thiz);
        thiz->shutdown = 1;
        return 0;
    } else if(events & ALOOPER_EVENT_OUTPUT) {
        //app_sys_debug("%s handler write", __func__);
        rocky_connection_handle_write(thiz);
    } else {
        app_sys_debug("%s handler unkown event %d", __func__, events);
    }
    return 1;
}

void rocky_connection_shutdown(RockyConnection *thiz) {
    if(thiz != NULL) {
        thiz->shutdown = 1;
        ALooper_wake(thiz->client_looper);
        sem_wait(&thiz->sem);
    }
}

int rocky_connection_loop(RockyConnection *thiz) {
    if(thiz == NULL || thiz->fd < 0) {
        app_sys_error("client has not setup");
        return -1;
    }

    thiz->client_looper = ALooper_prepare(0);
    ALooper_acquire(thiz->client_looper);
    set_non_block(thiz->fd);
    ALooper_addFd(thiz->client_looper, thiz->fd, 0, ALOOPER_EVENT_INPUT,
                  (ALooper_callbackFunc)rocky_connection_event_handler, thiz);
    while(!thiz->shutdown) {
        ALooper_pollOnce(-1, NULL, NULL, NULL);
    }
    ALooper_removeFd(thiz->client_looper, thiz->fd);
    ALooper_release(thiz->client_looper);
    close(thiz->fd);
    sem_post(&thiz->sem);
}

void rocky_connection_destroy(RockyConnection *thiz) {
    if(thiz != NULL) {
        if(thiz->read_record_stream != NULL) {
            record_stream_free(thiz->read_record_stream);
        }
        thiz->read_record_stream = NULL;
        if(thiz->write_data_queue != NULL) {
            work_item_queue_destroy(thiz->write_data_queue);
        }
        thiz->write_data_queue = NULL;
        sem_destroy(&thiz->sem);
        free(thiz);
    }
}

//=================below functions are only used in server======================

void* rocky_connectoin_handler_callback_for_server_thread(void *arg, void *ctx) {
    RockyConnection* thiz = (RockyConnection *)arg;
    rocky_connection_loop(thiz);
}

void rocky_connection_destroy_callback_for_server_thread(void *arg, void *ctx) {
    rocky_connection_destroy((RockyConnection *)arg);
}